bottle 是一个高效、简洁、轻量的 web 框架,整个框架只有一个 bottle.py 文件,最新版本 0.12.10 只有 3751行。只有一个文件,没有其他依赖,这一点就已经很酷了。它和 Flask 很像,同样是用装饰器来表现路由,同样都是微框架,那么搞定 bottle.py 有意义吗,直接看 Flask 源代码不好吗?当然有意义,Flask 基于 Werkzeug(WSGI 工具集) 和 jinja2(模板引擎), 代码量不多,很优雅,扩展性也很强,但和单文件的 bottle.py 相比就显得厚重了。从 bootle.py 的源代码中,我们可以学到的是 Web 框架的核心是什么,掌握了核心思想,再去处理一些 Web 开发过程中的一些边边角角的问题就只需要利用经验了。
Web 应用框架
首先我们应该对 Web 应用框架有一个简单的了解。在我看来,基本所有的 Web 应用框架都做了一件最基本的事情:接受 HTTP 请求,(这个请求可能是 GET,POST,PUT等等)接收到请求后,Server 端进行一些处理,返回给客户端一个回应,就这么简单。一般来说,Web 应用框架,还会将一些通用的功能集成在一起,比如数据库访问接口、模板、会话管理,异常处理等等,提高代码的复用性,减轻 Web 开发时程序员的工作负荷。
bottlepy 概述
『Bottle 是一个快速,简单,轻量级的 Python WSGI Web 框架。单一文件,只依赖 Python 标准库』
def _cast(self, out, peek=None): if not out: if 'Content-Length' not in response: response['Content-Length'] = 0 return [] if isinstance(out, (tuple, list))\ and isinstance(out[0], (bytes, unicode)): out = out[0][0:0].join(out) # b'abc'[0:0] -> b'' if isinstance(out, unicode): out = out.encode(response.charset) if isinstance(out, bytes): if 'Content-Length' not in response: response['Content-Length'] = len(out) return [out]
if isinstance(out, HTTPError): out.apply(response) out = self.error_handler.get(out.status_code, self.default_error_handler)(out) return self._cast(out) if isinstance(out, HTTPResponse): out.apply(response) return self._cast(out.body)
if hasattr(out, 'read'): if 'wsgi.file_wrapper' in request.environ: return request.environ['wsgi.file_wrapper'](out) elif hasattr(out, 'close') or not hasattr(out, '__iter__'): return WSGIFileWrapper(out)
try: iout = iter(out) first = next(iout) while not first: first = next(iout) except StopIteration: return self._cast('') except HTTPResponse: first = _e() except (KeyboardInterrupt, SystemExit, MemoryError): raise except Exception: if not self.catchall: raise first = HTTPError(500, 'Unhandled exception', _e(), format_exc())
for method in methods: if method in self.static and path in self.static[method]: target, getargs = self.static[method][path] return target, getargs(path) if getargs else {} elif method in self.dyna_regexes: for combined, rules in self.dyna_regexes[method]: match = combined(path) if match: target, getargs = rules[match.lastindex - 1] return target, getargs(path) if getargs else {}
allowed = set([]) nocheck = set(methods) for method in set(self.static) - nocheck: if path in self.static[method]: allowed.add(verb) for method in set(self.dyna_regexes) - allowed - nocheck: for combined, rules in self.dyna_regexes[method]: match = combined(path) if match: allowed.add(method) if allowed: allow_header = ",".join(sorted(allowed)) raise HTTPError(405, "Method not allowed.", Allow=allow_header)
raise HTTPError(404, "Not found: " + repr(path))
首先从 environ 变量中拿到 http 方法和请求的路径,首先匹配注册的静态路由,匹配不成功,接着匹配动态路由,调用 dyna_regexes 中已经编译好的正则表达式进行匹配。如果匹配上了,返回该路径对应的路由以及相应的参数,如果还是没有匹配上,匹配相同路径的其他方法,如果匹配上了,返回 405 Method not allowed 错误,否则返回 404 Not found 错误。成功匹配后,通过 route.call(**args) 进行调用。
def __get__(self, obj, cls): if obj is None: return self key, storage = self.key, getattr(obj, self.attr) if key not in storage: storage[key] = self.getter(obj) return storage[key]
def __set__(self, obj, value): if self.read_only: raise AttributeError("Read-Only property.") getattr(obj, self.attr)[self.key] = value
def __delete__(self, obj): if self.read_only: raise AttributeError("Read-Only property.") del getattr(obj, self.attr)[self.key]
当我们使用 @DictProperty('environ', 'bottle.request.headers', read_only=True) 这样的装饰器时,被装饰的函数的返回值实际在 self.environ['bottle.request.headers'] 中,而且可以确保这个属性只是可读的。 还记得 LocalRequest 中的 environ = local_property() 这行,它确保了 environ 是线程安全的。
1 2 3 4 5 6 7 8 9 10
def local_property(name=None): if name: depr('local_property() is deprecated and will be removed.') #0.12 ls = threading.local() def fget(self): try: return ls.var except AttributeError: raise RuntimeError("Request context not initialized.") def fset(self, value): ls.var = value def fdel(self): del ls.var return property(fget, fset, fdel, 'Thread-local property')
def template(*args, **kwargs): tpl = args[0] if args else None adapter = kwargs.pop('template_adapter', SimpleTemplate) lookup = kwargs.pop('template_lookup', TEMPLATE_PATH) tplid = (id(lookup), tpl) if tplid not in TEMPLATES or DEBUG: settings = kwargs.pop('template_settings', {}) if isinstance(tpl, adapter): TEMPLATES[tplid] = tpl if settings: TEMPLATES[tplid].prepare(**settings) elif "\n" in tpl or "{" in tpl or "%" in tpl or '$' in tpl: TEMPLATES[tplid] = adapter(source=tpl, lookup=lookup, **settings) else: TEMPLATES[tplid] = adapter(name=tpl, lookup=lookup, **settings) if not TEMPLATES[tplid]: abort(500, 'Template (%s) not found' % tpl) for dictarg in args[1:]: kwargs.update(dictarg) return TEMPLATES[tplid].render(kwargs)